package pers.cz.logtail.handler;

import pers.cz.config.JefConfiguration;
import pers.cz.logtail.config.LogClientConfigItem;
import pers.cz.netty.client.NettyClient;
import pers.cz.netty.client.TcpNettyClient;
import pers.cz.netty.config.NettyConfigItem;
import pers.cz.netty.protocal.Message;
import pers.cz.netty.protocal.MsgType;
import pers.cz.util.FileUtils;
import pers.cz.util.NetUtils;

/**
 * @program: JefLogTail
 * @description: 文件消费者
 * @author: Cheng Zhi
 * @create: 2023-03-24 13:32
 **/
public class FileLineConsumer extends Thread{

    private NettyClient nettyClient;

    /**
     * 启动消费者
     */
    public FileLineConsumer(NettyClient nettyClient) {
        this.nettyClient = nettyClient;
    }

    public void run() {

        while (true) {
            try {
                Object take = FileLineNodeQueue.take();
                if (take instanceof FileLineNote) {
                    long startPoint = ((FileLineNote) take).getStartLine();
                    long endPoint = ((FileLineNote) take).getEndLine();
                    int blockSize = JefConfiguration.getInt(LogClientConfigItem.LOG_MESSAGE_SIZE, 1024 * 128);
                    while (startPoint < endPoint) {
                        String s = FileUtils.readBlock(((FileLineNote) take).getFile(), startPoint, blockSize );
                        Message message = new Message();
                        // 设置服务端ip和port,如果使用udp通信的时候会用到
                        message.setIp(JefConfiguration.get(NettyConfigItem.NETTY_SERVER_IP));
                        message.setPort(JefConfiguration.getInt(NettyConfigItem.NETTY_SERVER_PORT, 9999));
                        message.setMsgType(MsgType.DATA);
                        message.addAttachment("ip", NetUtils.getLocalIp());
                        message.addAttachment("fileName", ((FileLineNote) take).getFileName());
                        message.setContent(s);
                        nettyClient.send(message);
                        startPoint = startPoint + blockSize;
                    }
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            }

        }

    }
}
